package com.fwmagic.flink.projectcase.queryactivitycase.async;

import com.alibaba.fastjson.JSONPath;
import com.fwmagic.flink.projectcase.queryactivitycase.bean.ActivityBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

public class AsyncGPSToActivityBeanFunction extends RichAsyncFunction<String, ActivityBean> {

    private transient CloseableHttpAsyncClient httpclient = null;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        //初始化异步的HttpClient
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();

        httpclient = HttpAsyncClients.custom()
        .setMaxConnTotal(20).setDefaultRequestConfig(requestConfig).build();

        httpclient.start();
    }

    @Override
    public void asyncInvoke(String line, ResultFuture<ActivityBean> resultFuture) throws Exception {
        //u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
        String[] fields = line.split(",");
        String uid = fields[0];
        String aid = fields[1];
        String time = fields[2];
        int eventType = Integer.parseInt(fields[3]);
        double longitude = Double.parseDouble(fields[4]);
        double latitude = Double.parseDouble(fields[5]);


        String url = "https://restapi.amap.com/v3/geocode/regeo?output=json&location=" + longitude + "," + latitude + "&key=0cab9004c4e4c4c81966b2d85564cf96";
        //调用高德地图gps的restapi查询接口
        HttpGet httpGet = new HttpGet(url);
        Future<HttpResponse> future = httpclient.execute(httpGet, null);
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    String data = null;
                    HttpResponse response = future.get();
                    if (response != null && response.getStatusLine() != null
                            && response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                        //获取请求的json串
                        String result = EntityUtils.toString(response.getEntity());
                        //获取城市
                        //String city = JSONPath.read(result, "$.regeocode.addressComponent.city").toString();
                        System.err.println("result:" + result);
                        //获取省份
                        data = JSONPath.read(result, "$.regeocode.addressComponent.province").toString();
                    }
                    return data;
                } catch (Exception e) {
                    // Normally handled explicitly.
                    return null;
                }
            }
        }).thenAccept((String province) -> {
            resultFuture.complete(Collections.singleton(ActivityBean.of(uid, aid, "", time, eventType, province, longitude, latitude)));
        });


    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
